package com.wing.consumer;

import com.alibaba.fastjson.JSONObject;
import com.tencentcloudapi.tdmq.v20200217.models.CreateSubscriptionRequest;
import com.tencentcloudapi.tdmq.v20200217.models.CreateSubscriptionResponse;
import com.wing.config.BaseDataInit;
import com.wing.config.PulsarClientInit;
import com.wing.web.db.config.util.DynamicDataSource;
import com.wing.order.model.entity.CabInventoryProduct;
import com.wing.order.service.CabInventoryProductService;
import com.wing.util.PulsarCommons;
import com.wing.util.TencentCloudApiUtil;
import com.wing.web.utils.JwtUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

/**
 * 柜子MQ数据同步 (消费端)
 */
@Slf4j
@Component
public class CabInventoryProductConsumer {

    @Value("${pulsar.client.subscriptionName}")
    private String subscriptionName;

    @Value("${pulsar.client.topicUrl}")
    private String topicUrl;
    @Value("${pulsar.client.namespace}")
    private String namespace;
    @Value("${pulsar.client.clusterId}")
    private String clusterId;

    @Autowired
    private BaseDataInit baseDataInit;
    @Autowired
    private CabInventoryProductService cabInventoryProductService;



    @Autowired
    private TencentCloudApiUtil tencentCloudApiUtil;

    static PulsarClientInit pulsarclientinitultimate =null;
    public CabInventoryProductConsumer(PulsarClientInit pulsarclientinitultimate) {
        CabInventoryProductConsumer.pulsarclientinitultimate = pulsarclientinitultimate;
    }

    @PostConstruct
    @SuppressWarnings("unchecked")
    public void init(){
        DynamicDataSource.name.set(baseDataInit.ENTERPRISE_DATA.getCode()); // 切换数据源
        String subscriptionName = baseDataInit.ENTERPRISE_DATA.getCode()+"_"+PulsarCommons.CAB_INVENTORY_PRODUCT;
        CreateSubscriptionRequest req = new CreateSubscriptionRequest();
        req.setEnvironmentId("company_" + baseDataInit.ENTERPRISE_DATA.getCode());
        req.setTopicName(baseDataInit.ENTERPRISE_DATA.getCode() + "_"+PulsarCommons.CAB_INVENTORY_PRODUCT);
        req.setSubscriptionName(subscriptionName);
        req.setIsIdempotent(true);
        req.setClusterId(clusterId.substring(0,clusterId.length()-1));
        req.setAutoCreatePolicyTopic(true);
        CreateSubscriptionResponse resp = tencentCloudApiUtil.createSubscription(req);
        if(null == resp || !resp.getResult()){
            log.error("创建订阅关系失败" );
            return;
        }
        PulsarClient client = pulsarclientinitultimate.getPulsars();
        try {
            String url = topicUrl.split("/")[0] + "/company_" + baseDataInit.ENTERPRISE_DATA.getCode()+"/";
            Consumer<byte[]> subscribe = client.newConsumer()
                    .subscriptionName(subscriptionName)
                    .topic(url+baseDataInit.ENTERPRISE_DATA.getCode() + "_"+PulsarCommons.CAB_INVENTORY_PRODUCT)
                    .consumerName(subscriptionName)
                    // 配置从最早开始消费，否则可能会消费不到历史消息
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                    // 设置消息处理失败后，60秒后再此收到消息。
                    .enableRetry(true)
                    .negativeAckRedeliveryDelay(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)//仅共享消费模式支持重试和死信
                    .deadLetterPolicy(DeadLetterPolicy.builder()
                            //1.当消息消费异常时,消息保存到重试队列,进行重试;
                            //2.最大重试6次后，进入到死信队列,可对死信队列单独进行订阅处理
                            //指定最大重试次数：
                            .maxRedeliverCount(6)
                            //指定重试队列
                            .retryLetterTopic(url+subscriptionName+"-RETRY")
                            //指定死信队列
                            .deadLetterTopic(url+subscriptionName+"-DLQ")
                            .build())
                    .messageListener((MessageListener<byte[]>) (consumer, message) -> {
                        log.info("consumer = " + message.getTopicName());
                        log.info("consumer = " +  message.getPublishTime());
                        log.info("consumer = " +  new String(message.getValue()));
                        processLogic(message.getValue());
                        try {
                            consumer.acknowledge(message);
                        } catch (PulsarClientException e) {
                            log.error("云仓库存消费确认异常:",e);
                        }
                    })
                    .subscribe();

        } catch (Exception e) {
            log.error("云仓库存消费异常:",e);
        }
    }

    public void processLogic(byte[] msg){
        String data = new String(msg);
        data = StringEscapeUtils.unescapeJavaScript(data);
        log.info(data);
        JSONObject jsonObject = JSONObject.parseObject(data.substring(1,data.length()-1));
        //更新或添加

            //入库逻辑
        CabInventoryProduct platformInventory = JSONObject.parseObject(JSONObject.toJSON(jsonObject.get("data")).toString(), CabInventoryProduct.class);
        platformInventory.setVersion(null);
        CabInventoryProduct inventory = cabInventoryProductService.getInventoryProductVOBySkuCode(platformInventory.getSkuCode(), platformInventory.getWarehouseId(),1);
        if(inventory == null){
            CabInventoryProduct cabInventoryProduct = new CabInventoryProduct();
            cabInventoryProduct.setWarehouseId(platformInventory.getWarehouseId())
                    .setQuantity(platformInventory.getQuantity())
                    .setSkuCode(platformInventory.getSkuCode())
                    .setSkuType(2)
                    .setSourceType(1);
            cabInventoryProductService.save(cabInventoryProduct);
        }else{
            inventory.setQuantity(platformInventory.getQuantity() + inventory.getQuantity());
            cabInventoryProductService.updateById(inventory);
        }

    }

}
